home *** CD-ROM | disk | FTP | other *** search
/ PC World Komputer 2010 April / PCWorld0410.iso / hity wydania / Ubuntu 9.10 PL / karmelkowy-koliberek-desktop-9.10-i386-PL.iso / casper / filesystem.squashfs / usr / share / pyshared / DistUpgrade / DistUpgradeAptCdrom.py < prev    next >
Encoding:
Python Source  |  2009-11-02  |  11.2 KB  |  289 lines
  1. # DistUpgradeAptCdrom.py 
  2. #  
  3. #  Copyright (c) 2008 Canonical
  4. #  
  5. #  Author: Michael Vogt <michael.vogt@ubuntu.com>
  6. # 
  7. #  This program is free software; you can redistribute it and/or 
  8. #  modify it under the terms of the GNU General Public License as 
  9. #  published by the Free Software Foundation; either version 2 of the
  10. #  License, or (at your option) any later version.
  11. # 
  12. #  This program is distributed in the hope that it will be useful,
  13. #  but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. #  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15. #  GNU General Public License for more details.
  16. # 
  17. #  You should have received a copy of the GNU General Public License
  18. #  along with this program; if not, write to the Free Software
  19. #  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  20. #  USA
  21.  
  22. import re
  23. import os
  24. import apt
  25. import apt_pkg
  26. import logging
  27. import gzip
  28. import shutil
  29. import subprocess
  30. from gettext import gettext as _
  31.  
  32.  
  33. class AptCdromError(Exception):
  34.     " base exception for apt cdrom errors "
  35.     pass
  36.  
  37. class AptCdrom(object):
  38.     " represents a apt cdrom object "
  39.  
  40.     def __init__(self, view, path):
  41.         self.view = view
  42.         self.cdrompath = path
  43.         # the directories we found on disk with signatures, packages and i18n
  44.         self.packages = set()
  45.         self.signatures = set()
  46.         self.i18n = set()
  47.  
  48.     def restoreBackup(self, backup_ext):
  49.         " restore the backup copy of the cdroms.list file (*not* sources.list)! "
  50.         cdromstate = os.path.join(apt_pkg.Config.FindDir("Dir::State"),
  51.                                   apt_pkg.Config.Find("Dir::State::cdroms"))
  52.         if os.path.exists(cdromstate+backup_ext):
  53.             shutil.copy(cdromstate+backup_ext, cdromstate)
  54.         # mvo: we don't have to care about restoring the sources.list here because
  55.         #      aptsources will do this for us anyway
  56.         
  57.  
  58.     def _scanCD(self):
  59.         """ 
  60.         scan the CD for interessting files and return them as:
  61.         (packagesfiles, signaturefiles, i18nfiles)
  62.         """
  63.         packages = set()
  64.         signatures = set()
  65.         i18n = set()
  66.         for root, dirs, files in os.walk(self.cdrompath, topdown=True):
  67.             if (root.endswith("debian-installer") or 
  68.                 root.endswith("dist-upgrader")):
  69.                 del dirs[:]
  70.                 continue
  71.             elif  ".aptignr" in files:
  72.                 continue
  73.             elif "Packages" in files:
  74.                 packages.add(os.path.join(root,"Packages"))
  75.             elif "Packages.gz" in files:
  76.                 packages.add(os.path.join(root,"Packages.gz"))
  77.             elif "Sources" in files or "Sources.gz" in files:
  78.                 logging.error("Sources entry found in %s but not supported" % root)
  79.             elif "Release.gpg" in files:
  80.                 signatures.add(os.path.join(root,"Release.gpg"))
  81.             elif "i18n" in dirs:
  82.                 for f in os.listdir(os.path.join(root,"i18n")):
  83.                     i18n.add(os.path.join(root,"i18n",f))
  84.             # there is nothing under pool but deb packages (no
  85.             # indexfiles, so we skip that here
  86.             elif os.path.split(root)[1] == ("pool"):
  87.                 del dirs[:]
  88.         return (packages, signatures, i18n)
  89.  
  90.     def _writeDatabase(self):
  91.         " update apts cdrom.list "
  92.         dbfile = apt_pkg.Config.FindFile("Dir::State::cdroms")
  93.         cdrom = apt_pkg.GetCdrom()
  94.         (res,id)=cdrom.Ident(apt.progress.CdromProgress())
  95.         label = self._readDiskName()
  96.         out=open(dbfile,"a")
  97.         out.write('CD::%s "%s";\n' % (id, label))
  98.         out.write('CD::%s::Label "%s";\n' % (id, label))
  99.  
  100.     def _dropArch(self, packages):
  101.         " drop architectures that are not ours "
  102.         # create a copy
  103.         packages = set(packages)
  104.         # now go over the packagesdirs and drop stuff that is not
  105.         # our binary-$arch 
  106.         arch = apt_pkg.Config.Find("APT::Architecture")
  107.         for d in set(packages):
  108.             if "/binary-" in d and not arch in d:
  109.                 packages.remove(d)
  110.         return packages
  111.     
  112.     def _readDiskName(self):
  113.         diskname = None
  114.         info = os.path.join(self.cdrompath, ".disk","info")
  115.         if os.path.exists(info):
  116.             diskname = open(info).read()
  117.             for special in ('"',']','[','_'):
  118.                 diskname = diskname.replace(special,'_')
  119.         return diskname
  120.  
  121.     def _generateSourcesListLine(self, diskname, packages):
  122.         # see apts indexcopy.cc:364 for details
  123.         path = ""                                    
  124.         dist = ""
  125.         comps = []
  126.         for d in packages:
  127.             # match(1) is the path, match(2) the dist
  128.             # and match(3) the components
  129.             m = re.match("(.*)/dists/([^/]*)/(.*)/binary-*", d)
  130.             if not m:
  131.                 raise AptCdromError, _("Could not calculate sources.list entry")
  132.             path = m.group(1)
  133.             dist = m.group(2)
  134.             comps.append(m.group(3))
  135.         if not path or not comps:
  136.             return None
  137.         comps.sort()
  138.         pentry = "deb cdrom:[%s]/ %s %s" % (diskname, dist, " ".join(comps))
  139.         return pentry
  140.  
  141.     def _copyTranslations(self, translations, targetdir=None):
  142.         if not targetdir:
  143.             targetdir=apt_pkg.Config.FindDir("Dir::State::lists")
  144.         diskname = self._readDiskName()
  145.         for f in translations:
  146.             fname = apt_pkg.URItoFileName("cdrom:[%s]/%s" % (diskname,f[f.find("dists"):]))
  147.             outf = os.path.join(targetdir,os.path.splitext(fname)[0])
  148.             if f.endswith(".gz"):
  149.                 g=gzip.open(f)
  150.                 out=open(outf,"w")
  151.                 # uncompress in 64k chunks
  152.                 while True:
  153.                     s=g.read(64000)
  154.                     out.write(s)
  155.                     if s == "":
  156.                         break
  157.             else:
  158.                 shutil.copy(f,outf)
  159.         return True
  160.  
  161.     def _copyPackages(self, packages, targetdir=None):
  162.         if not targetdir:
  163.             targetdir=apt_pkg.Config.FindDir("Dir::State::lists")
  164.         # CopyPackages()
  165.         diskname = self._readDiskName()
  166.         for f in packages:
  167.             fname = apt_pkg.URItoFileName("cdrom:[%s]/%s" % (diskname,f[f.find("dists"):]))
  168.             outf = os.path.join(targetdir,os.path.splitext(fname)[0])
  169.             if f.endswith(".gz"):
  170.                 g=gzip.open(f)
  171.                 out=open(outf,"w")
  172.                 # uncompress in 64k chunks
  173.                 while True:
  174.                     s=g.read(64000)
  175.                     out.write(s)
  176.                     if s == "":
  177.                         break
  178.             else:
  179.                 shutil.copy(f,outf)
  180.         return True
  181.  
  182.     def _verifyRelease(self, signatures):
  183.         " verify the signatues and hashes "
  184.         gpgv = apt_pkg.Config.Find("Dir::Bin::gpg","/usr/bin/gpgv")
  185.         keyring = apt_pkg.Config.Find("Apt::GPGV::TrustedKeyring",
  186.                                       "/etc/apt/trusted.gpg")
  187.         for sig in signatures:
  188.             basepath = os.path.split(sig)[0]
  189.             # do gpg checking
  190.             releasef = os.path.splitext(sig)[0]
  191.             cmd = [gpgv,"--keyring",keyring,
  192.                    "--ignore-time-conflict",
  193.                    sig, releasef]
  194.             ret = subprocess.call(cmd)
  195.             if not (ret == 0):
  196.                 return False
  197.             # now do the hash sum checks
  198.             t=apt_pkg.ParseTagFile(open(releasef))
  199.             t.Step()
  200.             for entry in t.Section["SHA256"].split("\n"):
  201.                 (hash,size,name) = entry.split()
  202.                 f=os.path.join(basepath,name)
  203.                 if not os.path.exists(f):
  204.                     logging.info("ignoring missing '%s'" % f)
  205.                     continue
  206.                 sum = apt_pkg.sha256sum(open(f))
  207.                 if not (sum == hash):
  208.                     logging.error("hash sum mismatch expected %s but got %s" % (hash, sum))
  209.                     return False
  210.         return True
  211.  
  212.     def _copyRelease(self, signatures, targetdir=None):
  213.         " copy the release file "
  214.         if not targetdir:
  215.             targetdir=apt_pkg.Config.FindDir("Dir::State::lists")
  216.         diskname = self._readDiskName()
  217.         for sig in signatures:
  218.             releasef = os.path.splitext(sig)[0]
  219.             # copy both Release and Release.gpg
  220.             for f in (sig, releasef):
  221.                 fname = apt_pkg.URItoFileName("cdrom:[%s]/%s" % (diskname,f[f.find("dists"):]))
  222.                 shutil.copy(f,os.path.join(targetdir,fname))
  223.         return True
  224.  
  225.     def _doAdd(self):
  226.         " reimplement pkgCdrom::Add() in python "
  227.         # os.walk() will not follow symlinks so we don't need
  228.         # pkgCdrom::Score() and not dropRepeats() that deal with
  229.         # killing the links
  230.         (self.packages, self.signatures, self.i18n) = self._scanCD()
  231.         self.packages = self._dropArch(self.packages)
  232.         if len(self.packages) == 0:
  233.             logging.error("no useable indexes found on CD, wrong ARCH?")
  234.             raise AptCdromError, _("Unable to locate any package files, perhaps this is not a Ubuntu Disc or the wrong architecture?")
  235.  
  236.         # CopyAndVerify
  237.         if self._verifyRelease(self.signatures):
  238.             self._copyRelease(self.signatures)
  239.         
  240.         # copy the packages and translations
  241.         self._copyPackages(self.packages)
  242.         self._copyTranslations(self.i18n)
  243.  
  244.         # add CD to cdroms.list "database" and update sources.list
  245.         diskname = self._readDiskName()
  246.         if not diskname:
  247.             logging.error("no .disk/ directory found")
  248.             return False
  249.         debline = self._generateSourcesListLine(diskname, self.packages)
  250.         
  251.         # prepend to the sources.list
  252.         sourceslist=apt_pkg.Config.FindFile("Dir::Etc::sourcelist")
  253.         content=open(sourceslist).read()
  254.         open(sourceslist,"w").write("# added by the release upgrader\n%s\n%s" % (debline,content))
  255.         self._writeDatabase()
  256.  
  257.         return True
  258.  
  259.     def add(self, backup_ext=None):
  260.         " add a cdrom to apt's database "
  261.         logging.debug("AptCdrom.add() called with '%s'", self.cdrompath)
  262.         # do backup (if needed) of the cdroms.list file
  263.         if backup_ext:
  264.             cdromstate = os.path.join(apt_pkg.Config.FindDir("Dir::State"),
  265.                                       apt_pkg.Config.Find("Dir::State::cdroms"))
  266.             if os.path.exists(cdromstate):
  267.                 shutil.copy(cdromstate, cdromstate+backup_ext)
  268.         # do the actual work
  269.         apt_pkg.Config.Set("Acquire::cdrom::mount",self.cdrompath)
  270.         apt_pkg.Config.Set("APT::CDROM::NoMount","true")
  271.         # FIXME: add cdrom progress here for the view
  272.         progress = self.view.getCdromProgress()
  273.         try:
  274.             res = self._doAdd()
  275.         except (SystemError, AptCdromError), e:
  276.             logging.error("can't add cdrom: %s" % e)
  277.             self.view.error(_("Failed to add the CD"),
  278.                              _("There was a error adding the CD, the "
  279.                                "upgrade will abort. Please report this as "
  280.                                "a bug if this is a valid Ubuntu CD.\n\n"
  281.                                "The error message was:\n'%s'") % e)
  282.             return False
  283.         logging.debug("AptCdrom.add() returned: %s" % res)
  284.         return res
  285.  
  286.     def __nonzero__(self):
  287.         """ helper to use this as 'if cdrom:' """
  288.         return self.cdrompath is not None
  289.